package com.winit.schedule.job.tracker.remoting;

import io.netty.channel.ChannelHandlerContext;

import java.text.ParseException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.event.Dispatcher;
import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.protocol.Protocol;
import com.winit.schedule.core.protocol.command.CompleteTaskRequest;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.constants.TaskStatus;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.repository.entity.TaskRunnerEntity;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.core.utils.DateUtils;
import com.winit.schedule.job.tracker.event.TaskEventType;
import com.winit.schedule.job.tracker.event.TaskStatusChangeEvent;
import com.winit.schedule.job.tracker.event.persist.DataPersistEvent;
import com.winit.schedule.job.tracker.event.persist.DataPersistEventType;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;
import com.winit.schedule.remoting.exception.RemotingCommandException;
import com.winit.schedule.remoting.protocol.RemotingCommand;

/**
 * 完成任务处理器
 * 
 * @author jianke.zhang 2015-3-24 上午11:15:24
 * @since 1.0
 */
public class CompleteTaskProcessor extends AbstractProcessor {

    private final static Logger LOGGER = LoggerFactory.getLogger(CompleteTaskProcessor.class);

    private ScheduleDataManager scheduleDataManager;

    private Dispatcher          dataDispatcher;

    public CompleteTaskProcessor(RemotingService remotingService){
        super(remotingService);
        this.dataDispatcher = remotingService.getJobTrackerContext().getDataDispatcher();
        this.scheduleDataManager = remotingService.getJobTrackerContext().getScheduleDataManager();
    }

    @SuppressWarnings("unchecked")
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
                                                                                             throws RemotingCommandException {
        CompleteTaskRequest body = request.getBody();
        String taskRunnerUuid = body.getTaskRunner().getTaskRunnerUuid();
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("From " + taskRunnerUuid + ", task execute " + body.getTaskStatus() + ", taskItems : "
                         + body.getTask().getTaskItems());
        }

        if (TaskStatus.COMPLETE.equals(body.getTaskStatus())) {

            long nextRunTime = 0L;
            try {
                nextRunTime = DateUtils.getNextValidTimeAfter4Cron(body.getTask().getRunStartTime(),
                    Clock.getCurrentDate());
            } catch (ParseException e) {
                LOGGER.warn("getNextValidTimeAfter4Cron : " + body.getTask().getRunStartTime(), e);
            }

            updateTaskItemStatus(body.getTask(), body.getTaskRunner(), TaskItemStatus.COMPLETE);
            dataDispatcher.getEventHandler().handle(new DataPersistEvent(DataPersistEventType.TASKITEM_STATUS_UPDATE,
                body.getTaskRunner(),
                body.getTask().getTaskItems(),
                TaskItemStatus.COMPLETE));

            // 无法获取下一次执行时间则表示任务已经执行完成
            if (nextRunTime == 0) {
                scheduleDataManager.completeTask(body.getTaskRunner(), body.getTask());
            }

            // 日志记录：任务项执行完成
            this.remotingService.getJobTrackerContext()
                .getDispatcher()
                .getEventHandler()
                .handle(new TaskStatusChangeEvent(TaskEventType.TASK_COMPLETE, body.getTask(), taskRunnerUuid));
        } else if (TaskStatus.STOP.equals(body.getTaskStatus())) {

            updateTaskItemStatus(body.getTask(), body.getTaskRunner(), TaskItemStatus.STOP);
            dataDispatcher.getEventHandler().handle(new DataPersistEvent(DataPersistEventType.TASKITEM_STATUS_UPDATE,
                body.getTaskRunner(),
                body.getTask().getTaskItems(),
                TaskItemStatus.STOP));

            // 日志记录：任务项被人工终止
            this.remotingService.getJobTrackerContext()
                .getDispatcher()
                .getEventHandler()
                .handle(new TaskStatusChangeEvent(TaskEventType.TASK_STOP, body.getTask(), taskRunnerUuid));
        } else if (TaskStatus.RUNNER_FAILURE.equals(body.getTaskStatus())) {
            // TaskTracker TaskRunner任务执行器异常，停止任务执行
            updateTaskItemStatus(body.getTask(), body.getTaskRunner(), TaskItemStatus.STOP);
            dataDispatcher.getEventHandler().handle(new DataPersistEvent(DataPersistEventType.TASKITEM_STATUS_UPDATE,
                body.getTaskRunner(),
                body.getTask().getTaskItems(),
                TaskItemStatus.STOP));

            // 日志记录：任务runner执行失败
            this.remotingService.getJobTrackerContext()
                .getDispatcher()
                .getEventHandler()
                .handle(new TaskStatusChangeEvent(TaskEventType.TASK_RUNNER_FAILURE,
                    body.getTask(),
                    taskRunnerUuid,
                    body.getException()));
        } else if (TaskStatus.EXECUTE_FAILURE.equals(body.getTaskStatus())) {
            // 任务接口执行失败不停止任务执行

            // 任务执行接口返回执行失败，
            this.remotingService.getJobTrackerContext()
                .getDispatcher()
                .getEventHandler()
                .handle(new TaskStatusChangeEvent(TaskEventType.TASK_EXE_FAILURE,
                    body.getTask(),
                    taskRunnerUuid,
                    body.getException()));
        }

        return RemotingCommand.createResponseCommand(Protocol.ResponseCode.SUCCESS.code(),
            "complete task received success!");
    }

    private void updateTaskItemStatus(AssignedTask task, TaskRunnerEntity runner, TaskItemStatus taskItemStatus) {
        if (null != task.getTaskItems() && !task.getTaskItems().isEmpty()) {
            // 每次心跳只有一个完成的任务项
            TaskItemEntity taskItem = task.getTaskItems().get(0);
            taskItem.setTaskRunner(runner);
            taskItem.setTaskItemStatus(taskItemStatus);
            scheduleDataManager.updateTaskItemStatusByTaskRunnerUuid(taskItem);
        }
    }
}
